package com.ebupt.migu.music.schedule;

import java.util.*;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.regex.Matcher;
import java.util.regex.Pattern;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.scheduling.annotation.Async;
import org.springframework.scheduling.annotation.EnableScheduling;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;

import com.ebupt.migu.music.common.constants.UserGroupConstants;
import com.ebupt.migu.music.common.utils.UpdateUserGroupSize;
import com.ebupt.migu.music.redis.service.RedisService;
import com.ebupt.migu.music.userGroup.mapper.NewCrowdMapper;
import com.google.gson.Gson;

import cn.hutool.http.HttpUtil;
import cn.hutool.json.JSONArray;
import cn.hutool.json.JSONObject;
import lombok.extern.slf4j.Slf4j;

/**
 * @author liuzz
 */
@Slf4j
@Component
@EnableScheduling
public class UserGroupTask {

    @Value("${spring.elasticsearch.hosts}")
    private String hosts;
    @Value("${spring.elasticsearch.port}")
    private String port;
    @Value("${spring.elasticsearch.alias}")
    private String alias;

    @Autowired
    private NewCrowdMapper newCrowdMapper;
    @Autowired
    private RedisService redisService;
    @Autowired
    private UpdateUserGroupSize updateUserGroupSize;

    static ExecutorService fixedThreadPool = Executors.newFixedThreadPool(8);

    /**
     * 人群重组
     */
    @Async("taskExecutor")
    @Scheduled(cron = "${cron.mergeGroup}")
    public void mergeGroup() {
        log.info("++++++++++++++++++++++++ mergeGroup");

        List<HashMap<String, Object>> crowdList = newCrowdMapper.queryMergeGroup();
        log.info("当前待更新的重组人群数量：{}", crowdList.size());
        Map<String, Object> params = new HashMap<>();

        for (Map<String, Object> crowd : crowdList) {

            String groupId = crowd.get("groupId").toString();
            String resGroups = crowd.get("resGroups").toString();
            log.info("groupId=" + groupId);
            log.info("res_groups=" + resGroups);

            // 先清除redis的数据
            redisService.remove_key(groupId);
            redisService.remove_key(groupId + "-uid");
            redisService.remove_key(groupId + "-imei");
            redisService.remove_key(groupId + "-phone");

            String[] idArr = resGroups.split(",");

            Map<String, Object> crowdMap1 = new HashMap<>();
            crowdMap1.put("groupId", idArr[0]);
            Map<String, Object> crowdForMerge1 = newCrowdMapper.queryByGroupID(crowdMap1);
            Map<String, Object> crowdMap2 = new HashMap<>();
            crowdMap2.put("groupId", idArr[1]);
            Map<String, Object> crowdForMerge2 = newCrowdMapper.queryByGroupID(crowdMap2);
            String masterIsStatic = String.valueOf(crowdForMerge1.get("isStaticCustomer"));
            String slaveIsStatic = String.valueOf(crowdForMerge2.get("isStaticCustomer"));
            if ("1".equals(masterIsStatic) || "1".equals(slaveIsStatic)) {
                log.info("包含静态客群的重组，由标签系统进行融合");
                continue;
            }
            String groupId1 = crowdForMerge1.get("groupId").toString();
            String groupId2 = crowdForMerge2.get("groupId").toString();
            String mergeType = crowd.get("createWay").toString();
            log.info("进行纯粹的动态客群重组- master：{}, slavel:{}", groupId1, groupId2);
            // 状态改为正在更新
            params.clear();
            params.put("createStatus", UserGroupConstants.UPDATING);
            params.put("groupId", groupId);
            newCrowdMapper.updateCrowd(params);
            // 存入redis
            boolean merge =
                setMergeGroupToRedis(groupId1, groupId2, groupId, mergeType, redisService, hosts, port, alias);
            log.info("存入redis结果：{}", merge);
            if (merge) {
                log.info("更新客群数据");
                updateUserGroupSize.update(groupId, 2, 0);
            }
            log.info("重组完成：{}", groupId);
        }
    }

    public static Map<String, Integer> setStaticGroupToRedis(String groupId, String hosts, String port, String index,
        String json, RedisService redisService) {
        // 解析json
        Gson gson = new Gson();
        Map<String, Object> groupAttributeMap = gson.fromJson(json, Map.class);

        // 组装查询条件
        List<String> comTypeList = null;
        List<String> ageList = null;
        List<String> areaNameList = null;
        List<String> userLevelList = null;
        List<String> chnIdList = null;
        List<String> phoneSystemList = null;

        String sex = "";
        String loginStatus = "";

        if (groupAttributeMap.get("comType") != null && !"".equals(groupAttributeMap.get("comType").toString())) {
            comTypeList = (List)groupAttributeMap.get("comType");
        }

        if (groupAttributeMap.get("age") != null && !"".equals(groupAttributeMap.get("age").toString())) {
            ageList = (List)groupAttributeMap.get("age");
        }

        if (groupAttributeMap.get("areaName") != null && !"".equals(groupAttributeMap.get("areaName").toString())) {
            areaNameList = (List)groupAttributeMap.get("areaName");
        }

        if (groupAttributeMap.get("userLevel") != null && !"".equals(groupAttributeMap.get("userLevel").toString())) {
            userLevelList = (List)groupAttributeMap.get("userLevel");
        }

        if (groupAttributeMap.get("chnId") != null && !"".equals(groupAttributeMap.get("chnId").toString())) {
            chnIdList = (List)groupAttributeMap.get("chnId");
        }

        if (groupAttributeMap.get("phoneSystem") != null
            && !"".equals(groupAttributeMap.get("phoneSystem").toString())) {
            phoneSystemList = (List)groupAttributeMap.get("phoneSystem");

            for (int i = 0; i < phoneSystemList.size(); i++) {
                if (phoneSystemList.get(i).equals("1")) {
                    phoneSystemList.set(i, "android");
                } else if (phoneSystemList.get(i).equals("1")) {
                    phoneSystemList.set(i, "ios");
                }
            }
        }

        if (groupAttributeMap.get("sex") != null && !"".equals(groupAttributeMap.get("sex").toString())) {
            sex = groupAttributeMap.get("sex").toString();
        }

        if (groupAttributeMap.get("loginStatus") != null
            && !"".equals(groupAttributeMap.get("loginStatus").toString())) {
            loginStatus = groupAttributeMap.get("loginStatus").toString();
        }

        String params = "";
        if (!((sex == null || "".equals(sex)) && (loginStatus == null || "".equals(loginStatus))
            && (comTypeList == null || comTypeList.size() == 0) && (ageList == null || ageList.size() == 0)
            && (areaNameList == null || areaNameList.size() == 0)
            && (userLevelList == null || userLevelList.size() == 0) && (chnIdList == null || chnIdList.size() == 0)
            && (phoneSystemList == null || phoneSystemList.size() == 0))) {
            sex = String.format("sex:(%s)", "%22" + groupAttributeMap.get("sex").toString() + "%22"); // "双引号转义 %22
            loginStatus =
                String.format("login_status:(%s)", "%22" + groupAttributeMap.get("loginStatus").toString() + "%22"); // "双引号转义
                                                                                                                     // %22

            String comType = String.format("com_type:(%s)", getParams(comTypeList));
            String age = String.format("age:(%s)", getParams(ageList));
            String areaName = String.format("area_name:(%s)", getParams(areaNameList));
            String userLevel = String.format("user_level:(%s)", getParams(userLevelList));
            String phoneSystem = String.format("phone_system:(%s)", getParams(phoneSystemList));
            String chnId = String.format("chn_id:(%s)", getParams(chnIdList));

            params = "(" + comType + "AND+" + sex + "AND+" + age + "AND+" + areaName + "AND+" + userLevel + "AND+"
                + loginStatus + "AND+" + chnId + "AND+" + phoneSystem + ")";
        }

        // 从es查询数据放到redis
        Integer userGroupSizeSum = 0;
        Integer clientUserSum = 0;
        Integer mobileUserSum = 0;
        Integer total = 0;
        Integer size = 10000;

        String result = "";
        String sort = "";
        JSONObject jsonObject;

        Map<String, Integer> retMap = new HashMap<>();
        Map<String, String> infoMap = new HashMap<>();
        infoMap.put("uidStr", "");
        infoMap.put("imeiStr", "");
        infoMap.put("msisdnStr", "");

        // 解析返回结果并存入redis
        String url =
            String.format("http://%s:%s/%s/_search?sort=row_id:asc&size=%s&q=%s", hosts, port, index, size, params);
        result = HttpUtil.get(url);
        // log.info("===================>>>>>>>>>>>>>>>> result="+result);
        jsonObject = new JSONObject(result);
        total = Integer.parseInt(jsonObject.getByPath("hits.total").toString());
        Integer page = total / size + 1;

        // 如果大于七百万则不执行
        if (total < 7000000) {
            // 第一次执行获取sort才能进行后续查询
            Map<String, Object> tmpMap = getEsSetAndAddToRedis(groupId, result, redisService);
            userGroupSizeSum += (Integer)tmpMap.get("userGroupSize");
            clientUserSum += (Integer)tmpMap.get("clientUser");
            mobileUserSum += (Integer)tmpMap.get("mobileUser");
            sort = tmpMap.get("sort").toString();

            // 如果没取完
            for (int i = 1; i < page; i++) {
                String body = String.format("{\"search_after\":[%s]}", sort);
                log.debug("===================>>>>>>>>>>>>>>>> search_after body=" + body);
                result = HttpUtil.post(url, body);

                tmpMap = getEsSetAndAddToRedis(groupId, result, redisService);
                userGroupSizeSum += (Integer)tmpMap.get("userGroupSize");
                clientUserSum += (Integer)tmpMap.get("clientUser");
                mobileUserSum += (Integer)tmpMap.get("mobileUser");
                sort = tmpMap.get("sort").toString();
            }
        }

        // 字符串放到redis
        String uidStr = infoMap.get("uidStr");
        String imeiStr = infoMap.get("imeiStr");
        String msisdnStr = infoMap.get("msisdnStr");
        if (!"".equals(uidStr)) {
            uidStr = uidStr.substring(0, uidStr.length() - 1);
            redisService.set(groupId + "-uid", uidStr);
        }

        if (!"".equals(imeiStr)) {
            imeiStr = imeiStr.substring(0, imeiStr.length() - 1);
            redisService.set(groupId + "-imei", imeiStr);
        }

        if (!"".equals(msisdnStr)) {
            msisdnStr = msisdnStr.substring(0, msisdnStr.length() - 1);
            redisService.set(groupId + "-phone", msisdnStr);
        }
        retMap.put("total", total);
        retMap.put("userGroupSize", userGroupSizeSum);
        retMap.put("clientUser", clientUserSum);
        retMap.put("mobileUser", mobileUserSum);

        return retMap;
    }

    public static Map<String, Object> getEsSetAndAddToRedis(String groupId, String json, RedisService redisService) {
        Map<String, Object> retMap = new HashMap<>();

        JSONObject jsonObject = new JSONObject(json);
        JSONArray hits = (JSONArray)jsonObject.getByPath("hits.hits");
        Integer total = Integer.parseInt(jsonObject.getByPath("hits.total").toString());
        Integer sort = 0;
        Integer size = hits.size();

        Set<String> clientUserSet = new HashSet<>();
        Set<String> mobileUserSet = new HashSet<>();
        Set msisdnSet = new HashSet<>();
        Set imeiSet = new HashSet<>();
        Set uidSet = new HashSet<>();

        Set set = new TreeSet<>();

        for (int i = 0; i < size; i++) {
            Object obj = hits.get(i);
            String uid = ((JSONObject)obj).getByPath("_source.uid").toString();
            String msisdn = ((JSONObject)obj).getByPath("_source.msisdn").toString();
            String idfa = ((JSONObject)obj).getByPath("_source.idfa").toString();
            String imei = ((JSONObject)obj).getByPath("_source.imei").toString();
            String clentFlag = ((JSONObject)obj).getByPath("_source.client_flag").toString();
            String loginStatus = ((JSONObject)obj).getByPath("_source.login_status").toString();

            if (!"".equals(imei)) {
                set.add(String.format("%s,%s,%s", uid, imei, msisdn));
                imeiSet.add(imei);
            } else {
                set.add(String.format("%s,%s,%s", uid, idfa, msisdn));
                imeiSet.add(idfa);
            }

            msisdnSet.add(msisdn);
            uidSet.add(uid);

            if ("1".equals(clentFlag)) { // 客户端 comType = 1 咪咕用户 2 QQ音乐用户 3 酷我音乐用户 4 酷狗音乐用户 5 网易云音乐用户 6 虾米音乐用户
                clientUserSet.add(uid);
            }
            if ("1".equals(loginStatus)) { // 手机端
                mobileUserSet.add(msisdn);
            }

            if (i == size - 1) {
                sort = (Integer)((JSONObject)obj).getByPath("_source.row_id");
            }
        }

        fixedThreadPool.submit(() -> {
            // 剔除空数据
            set.remove("");
            msisdnSet.remove("");
            uidSet.remove("");
            imeiSet.remove("");

            redisService.setSet(groupId + "-phone", msisdnSet);
            redisService.setSet(groupId + "-uid", uidSet);
            redisService.setSet(groupId + "-imei", imeiSet);
        });

        retMap.put("userGroupSize", size);
        retMap.put("clientUser", clientUserSet.size());
        retMap.put("mobileUser", mobileUserSet.size());
        retMap.put("total", total);
        retMap.put("sort", sort);

        return retMap;
    }

    public static String getParams(List<String> list) {
        String params = "";

        if (list != null && list.size() > 0) {
            for (String element : list) {
                params += "%22" + element + "%22 ";
            }
        }

        if (!"".equals(params)) {
            params = params.substring(0, params.length() - 1);
        } else {
            params = "%22%22";
        }

        return params;
    }

    public boolean setMergeGroupToRedis(String srouceGroupId1, String srouceGroupId2, String targetGroupId,
        String mergeType, RedisService redisService, String hosts, String port, String index) {
        Map<String, Object> map = new HashMap<>();
        map.put("createStatus", 6);
        map.put("remark", "人群规模超过700万");
        map.put("groupId", targetGroupId);
        Set uidSet = merge(srouceGroupId1 + "-uid", srouceGroupId2 + "-uid", mergeType, redisService);
        if (uidSet.size() > 7000000) {
            log.info("人群规模超过700万:{}", targetGroupId);
            newCrowdMapper.updateCrowd(map);
            return false;
        }
        Set imeiSet = merge(srouceGroupId1 + "-imei", srouceGroupId2 + "-imei", mergeType, redisService);
        if (imeiSet.size() > 7000000) {
            log.info("人群规模超过700万:{}", targetGroupId);
            newCrowdMapper.updateCrowd(map);
            return false;
        }
        Set msisdnSet = merge(srouceGroupId1 + "-phone", srouceGroupId2 + "-phone", mergeType, redisService);
        if (msisdnSet.size() > 7000000) {
            log.info("人群规模超过700万:{}", targetGroupId);
            newCrowdMapper.updateCrowd(map);
            return false;
        }
        log.info("人群规模正常:{}", targetGroupId);
        redisService.setSet(targetGroupId + "-uid", uidSet);
        redisService.setSet(targetGroupId + "-imei", imeiSet);
        redisService.setSet(targetGroupId + "-phone", msisdnSet);
        return true;
    }

    private static Set merge(String srouceGroupId1, String srouceGroupId2, String mergeType,
        RedisService redisService) {
        Set result = new HashSet();
        Set set1 = redisService.getSet(srouceGroupId1);
        log.info("{}在redis中的数据量：{}", srouceGroupId1, set1.size());
        Set set2 = redisService.getSet(srouceGroupId2);
        log.info("{}在redis中的数据量：{}", srouceGroupId2, set2.size());
        switch (mergeType) {
            // 4-交集创建
            case "4":
                result.addAll(set1);
                result.retainAll(set2);
                break;
            // 5-并集创建
            case "5":
                result.addAll(set1);
                result.addAll(set2);
                break;
            // 6-左差集创建
            case "6":
                // 先取并集
                result.addAll(set1);
                result.addAll(set2);

                // 差集（左）：两个重组用户群，以第一个用户群为目标用户群，将两个用户群中只要有一个用户标识符（UID、IMEI/IDFA、手机号）相同的用户剔除；
                result.removeAll(set2);
                break;
            // 7-右差集创建
            case "7":
                // 先取并集
                result.addAll(set1);
                result.addAll(set2);

                // 差集（右）：两个重组用户群，以第二个用户群为目标用户群，将两个用户群中只要有一个用户标识符（UID、IMEI/IDFA、手机号）相同的用户剔除；
                result.removeAll(set1);
                break;
            // 8-对称差集创建
            case "8":
                // 先取并集
                result.addAll(set1);
                result.addAll(set2);

                // 取交集
                Set tmpIntersectSet = new HashSet<>();
                tmpIntersectSet.addAll(set1);
                tmpIntersectSet.retainAll(set2);

                // 差集（对称差集）：两个重组用户群，以两个用户群为目标用户群，将两个用户群中只要有一个用户标识符（UID、IMEI/IDFA、手机号）相同的用户剔除；
                result.removeAll(tmpIntersectSet);

                break;
            default:
                break;
        }

        return result;
    }

    /**
     * 对手机号码进行格式校验
     *
     * @param msisdn
     * @return
     */
    private static boolean isMsisdn(String msisdn) {
        Pattern pattern = Pattern.compile("^1[0-9]{10}$");
        Matcher matcher = pattern.matcher(msisdn);
        return matcher.matches();
    }
}
